package org.yuanzheng.tableAndSql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table, Tumble}
import org.apache.flink.types.Row
import org.yuanzheng.source.StationLog

/*每隔5秒钟统计，每个基站的通话数量，假设数据是乱序，最多延时3秒（watermark）*/
object TestWindowByTableAPI {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //读取数据
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.10", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })
      //引入水位线，延迟触发
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
        override def extractTimestamp(element: StationLog): Long = {
          element.callTime
        }
      })

    //创建动态table并指定EventTime字段(使用.rowtime)
    val table: Table = tableEnv.fromDataStream(stream, 'sid, 'callOut, 'callIn, 'callTime.rowtime)

    //2种滚动 Tumble
    //table.window(Tumble.over("5.second").on("callTime").as("windowTime"))
    val result: Table = table.window(Tumble over 5.second on 'callTime as 'windowTime) //开窗
      .groupBy('windowTime, 'sid) //必须使用两个字段分组
      .select('sid, 'windowTime.start, 'windowTime.end, 'sid.count) //聚合计算
    //2种滑动 Slide
    //table.window(Slide.over(10.second).every(5.second).on("callTime").as("windowTime"))
    //table.window(Slide over 10.second every 5.second on 'callTime as 'windowTime)

    tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()
    streamEnv.execute()
  }
}
